package com.example.cloudapirabbitmq.direct;

import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;

import java.util.UUID;

@Configuration
@Slf4j
public class RabbitMQProducerConfig implements RabbitListenerConfigurer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        // 设置消息处理工厂（支持JSON序列化）
        registrar.setMessageHandlerMethodFactory(jsonMessageHandlerMethodFactory());
    }

    @Bean
    public MessageHandlerMethodFactory jsonMessageHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(jsonMessageConverter()); // 配置JSON消息转换器
        return factory;
    }

    @Bean
    public MappingJackson2MessageConverter jsonMessageConverter() {
        return new MappingJackson2MessageConverter();
    }

    @PostConstruct
    public void init() {
//        ConfirmCallback处理消息是否成功到达交换机的确认结果。
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息成功到达交换机，ID：{}", correlationData.getId());
            } else {
                log.error("消息未到达交换机，原因：{}", cause);
                log.error("correlationData：{}", correlationData.getId());
                // 可在此处重试或记录失败消息

                // 尝试重新发送消息（简单重试示例）
                try {
                    CorrelationData corr = new CorrelationData();
                    corr.setId(correlationData.getId());
                    rabbitTemplate.convertAndSend(
                            "xx",
                            //RabbitMQQueueConfig.ORDER_EXCHANGE,
                            "xx",
//                            RabbitMQQueueConfig.ORDER_ROUTING_KEY,
                            "11111111111111",
                            message -> {
                                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                                message.getMessageProperties().setCorrelationId(correlationData.getId());
                                message.getMessageProperties().setMessageId(correlationData.getId());
                                return message;
                            },
                            corr
                    );
                } catch (Exception e) {
                    log.error("重试发送失败，消息放弃", e);
                    // 可将消息存入死信队列或数据库
                }

            }
        });

//         发送失败回调（路由失败时触发）
//        ReturnCallback处理路由失败场景（如交换机不存在或路由键不匹配）。
        rabbitTemplate.setReturnsCallback(returned -> {
            log.error("消息路由失败，交换机：{}，路由键：{}，消息：{}",
                    returned.getExchange(), returned.getRoutingKey(), new String(returned.getMessage().getBody()));
            // 可在此处将失败消息存入数据库或死信队列

            // 将失败消息发送到死信队列
            CorrelationData corr = new CorrelationData();
            corr.setId("qwertyuiop123456789");
            rabbitTemplate.convertAndSend(
                    "dlx.exchange",
                    "dlx.routing.key",
                    returned.getMessage().getBody(),
                    msg -> {
                        msg.getMessageProperties().setMessageId(corr.getId());
                        msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                        return msg;
                    },
                    corr
            );
        });
    }

    // 发送消息方法（携带唯一ID和重试逻辑）
    public void sendMessage(String exchange, String routingKey, String message) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        MessagePostProcessor processor = msg -> {
            msg.getMessageProperties().setMessageId(correlationData.getId());
            msg.getMessageProperties().setCorrelationId(correlationData.getId());
//            通过 MessageDeliveryMode.PERSISTENT确保消息写入磁盘，防止 RabbitMQ 重启后丢失。
            msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 持久化消息
            return msg;
        };
        rabbitTemplate.convertAndSend(exchange, routingKey, message, processor, correlationData);
    }


}